前言

在前面的章節中，我們了解了 Tool 是 LangChain Agent 調用外部能力的核心。
接下來，我們以Qdrant 向量搜尋 為例，示範如何把RAG裡面的Retrieval包裝成一個可供 Agent 使用的「工具」。

🧱 工具化設計思路

先介紹一下今天範例會用到的原件如下表:

🧠 實作：簡單 Qdrant 檢索器

下面先提供Qdrant於Retrieval這部份的簡單範例，其他如啟動collection等，可以參考筆者於Day 16:向量資料庫進階分析: Qdrant的介紹。

from qdrant_client import QdrantClient
from FlagEmbedding import BGEM3FlagModel

class SimpleRetriever:
    """簡單檢索器"""
    
    def __init__(self, collection_name="GrassOwl"):
        self.client = QdrantClient("localhost", port=6333)
        self.collection_name = collection_name
        print("正在載入 BGE-M3 模型...")
        self.embedding_model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True)
        print("✅ 模型載入完成")
    
    def encode_query(self, query):
        """將查詢文本轉換為向量"""
        embeddings = self.embedding_model.encode([query])
        return embeddings['dense_vecs'][0].tolist()
    
    def search(self, query, top_k=5):
        """執行相似度搜尋"""
        query_vector = self.encode_query(query)
        search_results = self.client.search(
            collection_name=self.collection_name,
            query_vector=query_vector,
            limit=top_k,
            with_payload=True,
            with_vectors=False
        )
        contents = [r.payload.get('content', '') for r in search_results]
        return contents

⚙️ 封裝成 LangChain Tool

在 LangChain 中，Tool 是讓 LLM 具備可調用外部功能（Function Calling） 的核心。
透過包裝成 Tool，Agent 才能依據需求主動呼叫這些工具完成任務。

🔧基本結構

LangChain 的工具通常繼承自 BaseTool 或使用 StructuredTool.from_function() 包裝。
基本原則：

from langchain.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

class MyToolInput(BaseModel):
    """定義輸入參數格式（建議要有）"""
    query: str = Field(description="查詢內容")
    top_k: int = Field(default=5, description="返回結果數量")

class MyTool(BaseTool):
    name = "my_tool"
    description = "這個工具用來搜尋資料庫中的內容"
    args_schema: Type[BaseModel] = MyToolInput  # 定義參數格式

    def _run(self, query: str, top_k: int = 5) -> str:
        """同步執行邏輯"""
        return f"搜尋 {query}（Top {top_k}）的結果"

    async def _arun(self, query: str, top_k: int = 5) -> str:
        """異步執行邏輯（非必要）"""
        return self._run(query, top_k)

🧩Qdrant search範例

下面是必要的套件:

from langchain.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field
from qdrant_client import QdrantClient
from FlagEmbedding import BGEM3FlagModel

下面這段是Qdrant search工具的參數 schema，用來：

• 定義工具需要哪些輸入。
• 給 LLM 一個明確的「欄位描述」。
• 驗證輸入資料型別（類似 FastAPI 的 Pydantic schema）。
  這個部份不一定要有，但建議要有，能讓 LLM 明確知道如何填參數。

class QdrantSearchInput(BaseModel):
    """Qdrant 搜尋工具的輸入參數"""
    query: str = Field(description="搜尋查詢文本")
    top_k: int = Field(default=5, description="返回結果數量")

下面是把Qdrant search包裝成tool的範例:

class QdrantSearchTool(BaseTool):
    """Qdrant 搜尋工具"""
    name = "qdrant_search"
    description = "當用戶提出關於草鴞的問題，用來搜尋 GrassOwl collection 中的相關文檔"
    args_schema: Type[BaseModel] = QdrantSearchInput
    
    def __init__(self, collection_name="GrassOwl"):
        super().__init__()
        self.client = QdrantClient("localhost", port=6333)
        self.collection_name = collection_name
        print("正在載入 BGE-M3 模型...")
        self.embedding_model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True)
        print("✅ 模型載入完成")
    
    def _run(self, query: str, top_k: int = 5) -> str:
        """執行搜尋"""
        try:
            embeddings = self.embedding_model.encode([query])
            query_vector = embeddings['dense_vecs'][0].tolist()
            search_results = self.client.search(
                collection_name=self.collection_name,
                query_vector=query_vector,
                limit=top_k,
                with_payload=True,
                with_vectors=False
            )
            contents = []
            for i, result in enumerate(search_results, 1):
                content = result.payload.get('content', '')
                contents.append(f"結果 {i}: {content}")
            
            return "\n\n".join(contents) if contents else "沒有找到相關結果"
        except Exception as e:
            return f"搜尋失敗: {str(e)}"
    
    async def _arun(self, query: str, top_k: int = 5) -> str:
        return self._run(query, top_k)

🦜 建立 MRKL Agent 並測試

以下示範如何用 LangChain 建立一個 MRKL Agent，這邊我們使用筆者前面於Day 15: 在地端運行 LLM：Ollama、vLLM 與 llama.cpp 比較以及ollama安裝介紹，使用ollama安裝的地端的llama-3-taiwan-8b(筆者先前範例有寫錯下載方式，真抱歉)為驅動MRKL的LLM引擎，讓它透過自然語言自動調用 QdrantSearchTool。

from langchain.agents import initialize_agent, AgentType
from langchain_ollama.llms import OllamaLLM

# ✅ 初始化 LLM（使用 Ollama）
llm = OllamaLLM(model="cwchang/llama-3-taiwan-8b-instruct")

# ✅ 建立工具
qdrant_tool = QdrantSearchTool(collection_name="GrassOwl")

# ✅ 建立 Agent
agent = initialize_agent(
    tools=[qdrant_tool],
    llm=llm,
    agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# ✅ 測試 Agent 呼叫 Qdrant 搜尋
response = agent.run("請幫我搜尋有關草鴞棲地保育的相關資料。")
print(response)

小結

到了這邊，其實已經完成AgenticRAG了! 後面我們就拼裝起來做測試吧!!